#include "Spark/Net/EventLoop.h"
#include "Spark/Net/Poller.h"
#include "Spark/Net/TimerQueue.h"
#include "Spark/Net/Channel.h"
#include "Spark/Net/IgnoreSignal.h"
#include "Spark/Utils/Logger.h"
#include <thread>
#include <sys/eventfd.h>
#include <fcntl.h>
#include <unistd.h>
#include <assert.h>
#include <stdexcept>

static int createEventFD()
{
    int eventFD = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
    assert(eventFD >= 0);
    return eventFD;
}

thread_local EventLoop* tlLoopInThisThread = nullptr;
static const int kPollTimeoutMS = 10000;

static IgnoreSignal ign(SIGPIPE);

EventLoop::EventLoop()
    : threadID_(std::this_thread::get_id()),
      poller_(Poller::newPoller(this)),
      timerQueue_(new TimerQueue(this)),
      wakeupFD_(createEventFD()),
      wakeupChannel_(new Channel(this, wakeupFD_)),
      localLoop_(&tlLoopInThisThread)
{
    assert(tlLoopInThisThread == nullptr);
    tlLoopInThisThread = this;
    wakeupChannel_->setCallback(CBType::READ, [this] { wakeupRead(); });
    wakeupChannel_->addReadEvent();
}

EventLoop::~EventLoop()
{
    quit();
    assert(!looping_);
    tlLoopInThisThread = nullptr;
    ::close(wakeupFD_);
}

void EventLoop::assertInLoopThread()
{
    assert(threadID_ == std::this_thread::get_id());
}

void EventLoop::resetTimerQueue()
{
    assertInLoopThread();
    assert(!looping_);
    timerQueue_->reset();
}

bool EventLoop::isInLoopThread() const
{
    return threadID_ == std::this_thread::get_id();
}

void EventLoop::moveToCurrentThread()
{
    if (isRunning())
        throw std::runtime_error("Cannot move thread while running!!");
    if (isInLoopThread())
        return;
    if (tlLoopInThisThread)
        throw std::runtime_error(
            "There is already an EventLoop in this thread!!");
    *localLoop_ = nullptr;
    tlLoopInThisThread = this;
    localLoop_ = &tlLoopInThisThread;
    auto id = const_cast<std::thread::id*>(&threadID_);
    *id = std::this_thread::get_id();
}

void EventLoop::runInLoop(Task task)
{
    if (isInLoopThread())
        task();
    else
        queueInLoop(std::move(task));
}

void EventLoop::queueInLoop(Task task)
{
    taskQueue_.push(std::move(task));
    if (!isInLoopThread() || callingFunctors_)
        wakeup();
}

template <class P, class R>
TimerID EventLoop::runAfter(std::chrono::duration<P, R> delay, Task task)
{
    return runAt(TimePoint::Now() + delay, std::move(task));
}

// https://isocpp.org/wiki/faq/templates#separate-template-fn-defn-from-decl

template TimerID EventLoop::runAfter(
    TimeInterval<long double, std::ratio<1l, 1l>>,
    std::function<void()>);
template TimerID EventLoop::runAfter(TimeInterval<long, std::ratio<1l, 1l>>,
                                     std::function<void()>);
template TimerID EventLoop::runAfter(TimeInterval<long, std::milli>,
                                     std::function<void()>);
template TimerID EventLoop::runAfter(TimeInterval<long, std::micro>,
                                     std::function<void()>);
template TimerID EventLoop::runAfter(TimeInterval<long, std::nano>,
                                     std::function<void()>);

TimerID EventLoop::runAt(TimePoint when, Task task)
{
    return timerQueue_->addTimer(std::move(task), when, SysTime::duration{});
}

template <class P, class R>
TimerID EventLoop::runEvery(TimeInterval<P, R> interval, Task task)
{
    return timerQueue_->addTimer(std::move(task),
                                 TimePoint::Now() + interval,
                                 interval);
}

template TimerID EventLoop::runEvery(
    TimeInterval<long double, std::ratio<1l, 1l>>,
    std::function<void()>);
template TimerID EventLoop::runEvery(TimeInterval<long, std::ratio<1l, 1l>>,
                                     std::function<void()>);
template TimerID EventLoop::runEvery(TimeInterval<long, std::milli>,
                                     std::function<void()>);
template TimerID EventLoop::runEvery(TimeInterval<long, std::micro>,
                                     std::function<void()>);
template TimerID EventLoop::runEvery(TimeInterval<long, std::nano>,
                                     std::function<void()>);

void EventLoop::disableTimer(TimerID id)
{
    if (looping_ && timerQueue_)
        timerQueue_->disableTimer(id);
}

void EventLoop::updateChannel(Channel* ch)
{
    assert(ch->ownerLoop() == this);
    assertInLoopThread();
    poller_->updateChannel(ch);
}

void EventLoop::removeChannel(Channel* ch)
{
    assert(ch->ownerLoop() == this);
    assertInLoopThread();
    poller_->removeChannel(ch);
}

void EventLoop::quit()
{
    quit_.store(true);
    if (!isInLoopThread())
    {
        wakeup();
    }
}

void EventLoop::loop()
{
    assert(!looping_);
    assertInLoopThread();
    looping_ = true;
    quit_ = false;
    while (!quit_)
    {
        activeChannels_.clear();
        poller_->poll(kPollTimeoutMS, &activeChannels_);
        for (Channel* ch : activeChannels_)
        {
            currentHandlingChannel_ = ch;
            currentHandlingChannel_->handleEvent();
        }
        doPendingTasks();
    }
    looping_ = false;
    quit_ = true;
}

EventLoop* EventLoop::getEventLoopOfThisThread()
{
    return tlLoopInThisThread;
}

void EventLoop::wakeup()
{
    uint64_t optval = 1;
    ssize_t n = ::write(wakeupFD_, &optval, sizeof optval);
    if (n != sizeof optval)
    {
        LOG_ERROR << "EventLoop::wakeup() writes " << n
                  << " bytes instead of 8";
    }
}

void EventLoop::wakeupRead()
{
    uint64_t optval = 1;
    ssize_t n = ::read(wakeupFD_, &optval, sizeof optval);
    if (n != sizeof optval)
    {
        LOG_ERROR << "EventLoop::wakeupRead() reads " << n
                  << " bytes instead of 8";
    }
}

void EventLoop::doPendingTasks()
{
    callingFunctors_ = true;
    while (!taskQueue_.empty())
    {
        Task task;
        while (taskQueue_.pop(task))
        {
            task();
        }
    }
    callingFunctors_ = false;
}
